package com.citusdata.elven.feeder;

import com.citusdata.elven.feeder.FileFeederRunnable;
import com.citusdata.elven.feeder.HttpRpcDataFeeder;
import com.citusdata.elven.feeder.Log4jHelper;
import com.citusdata.elven.httpRpc.HttpRpcChannelManager;
import com.citusdata.elven.linearRoad.LinearRoadCoordinator;
import com.citusdata.elven.router.ReplicationGroupRouter;
import com.google.protobuf.Message;
import org.apache.log4j.Logger;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;


public class DistributedDataFeeder {
  static { Log4jHelper.setApplicationName(DistributedDataFeeder.class); } 
  private static final Logger logger = Logger.getLogger(DistributedDataFeeder.class);

  private static final String INPUT_EXTENSION  = ".in";
  private static final String OUTPUT_EXTENSION = ".out";
  private static final String MEMBERSHIP_FILE = "membership.properties";

  private static List baseFileNames(String inputDirectory) {
    List baseFileNames = new ArrayList();
    File inputPath = new File(inputDirectory);
    File[] inputFiles = inputPath.listFiles();

    for (int i = 0; i < inputFiles.length; ++i) {
      String currentFile = inputFiles[i].getName();
      int extensionIndex = currentFile.lastIndexOf(INPUT_EXTENSION);

      if ( (0 < extensionIndex) && (extensionIndex < currentFile.length()) ) {
        String baseFileName = currentFile.substring(0, extensionIndex);
        baseFileNames.add(baseFileName);
      } else {
        logger.warn("Skipping file: " + currentFile);
      }
    }
    
    return baseFileNames;
  }


  private static List inputStreams(String inputDirectory, List baseFileNames) 
    throws IOException {
    List inputStreams = new ArrayList();    

    for (int i = 0; i < baseFileNames.size(); ++i) {
      String directoryName = inputDirectory + File.separator;
      String fileName = directoryName + baseFileNames.get(i) + INPUT_EXTENSION;
      BufferedReader inputStream = new BufferedReader(new FileReader(fileName));
      inputStreams.add(inputStream);
    }

    return inputStreams;
  }

  
  private static List outputStreams(String outputDirectory, List baseFileNames)
    throws IOException { 
    List outputStreams = new ArrayList();    

    for (int i = 0; i < baseFileNames.size(); ++i) {
      String directoryName = outputDirectory + File.separator;
      String fileName = directoryName + baseFileNames.get(i) + OUTPUT_EXTENSION;
      PrintWriter outputStream = new PrintWriter(new FileWriter(fileName));
      outputStreams.add(outputStream);
    }

    return outputStreams;
  }


  public static void main(String[] args) {
    // DistributedDataFeeder is different from other feeders in that it takes
    // directories instead of files, as input. It then iterates over each file
    // in the input directory, and feeds requests from these files
    // simultaneously. 
    if (args.length < 1) {
      logger.fatal("Usage: DistributedDataFeeder cardata_directory [output_directory]");
      System.exit(-1);
    } else {
      logger.info("DistributedDataFeeder starting to feed data");
    }

    // Check that the number of server threads is higher than the number of
    // client threads.
    HttpRpcDataFeeder.checkHttpRpcConfig();

    // Initialize data structures used for feeding data into the distributed system
    ReplicationGroupRouter router = new ReplicationGroupRouter(MEMBERSHIP_FILE);
    HttpRpcChannelManager channelManager = new HttpRpcChannelManager();
    LinearRoadCoordinator coordinator = new LinearRoadCoordinator(router, channelManager);

    List inputStreams  = null;
    List outputStreams = null;

    try {
      String inputDirectory  = args[0];
      String outputDirectory = "/var/tmp";
      if (args.length == 2) {
        outputDirectory = args[1];
      }

      List baseFileNames = baseFileNames(inputDirectory);
      inputStreams  = inputStreams(inputDirectory, baseFileNames);
      outputStreams = outputStreams(outputDirectory, baseFileNames);

      int fileCount = inputStreams.size();
      ExecutorService feedExecutor = Executors.newFixedThreadPool(fileCount);

      for (int index = 0; index < fileCount; ++index) {
        BufferedReader inputStream = (BufferedReader) inputStreams.get(index);
        PrintWriter   outputStream = (PrintWriter) outputStreams.get(index);
        
        FileFeederRunnable task = new FileFeederRunnable(coordinator,
                                                         inputStream, outputStream);
        feedExecutor.execute(task);
      }

      // Wait until worker threads finish feeding data
      feedExecutor.shutdown();
      feedExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

    } catch (IOException ex) {
      logger.error("Caught exception when distributing data feeds", ex);

    } catch (InterruptedException ex) {
      logger.error("Caught exception when distributing data feeds", ex);

    } finally {
      logger.info("Finished sending requests, shutting down");

      channelManager.shutdown();  // Shutdown open connections to servers 

      try {
        int inputSize = (inputStreams == null) ? 0 : inputStreams.size();
        for (int index = 0; index < inputSize; ++index) {
          BufferedReader inputStream = (BufferedReader) inputStreams.get(index);
          PrintWriter   outputStream = (PrintWriter) outputStreams.get(index);
          
          if (inputStream != null) {
            inputStream.close();
          }
          if (outputStream != null) {
            outputStream.close();
          }
        }
      } catch (IOException ignore) { }
    }
  }
}
